package com.zwan.oam_rtc.imps;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.github.kevinsawicki.http.HttpRequest;
import com.zwan.oam_rtc.bean.WintcpSettings;
import com.zwan.oam_rtc.constants.Const;
import com.zwan.oam_rtc.utils.messages.MessageHead;
import com.zwan.oam_rtc.utils.messages.MsgStruct;
import com.zwan.oam_rtc.utils.messages.RtcmsResponse;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

@Slf4j
public class ClientHandler implements Runnable {

    private int port;
    private String ipaddress;
    private int peerport;
    private String peerip;
    private int localpno;
    private ByteOrder byteOrder;
    private int peerModule;
    // 创建只有一个线程的线程池，可以在指定延迟后执行任务
    private final ScheduledExecutorService checkSchedule = Executors.newSingleThreadScheduledExecutor();
    private final ScheduledExecutorService connSchedule = Executors.newSingleThreadScheduledExecutor();
    private final ScheduledExecutorService regDtmfSchedule = Executors.newSingleThreadScheduledExecutor();
    private ScheduledFuture<?> checkFuture = null;
    private ScheduledFuture<?> connFuture = null;
    private ScheduledFuture<?> regDtmfFuture = null;

    private Socket socket;
    private int reconnectcount = 0;
    private static final int MAX_RECONNECTCOUNT = 10;
    private static int CONNECT_TIMEOUT = 5000;

    public ClientHandler() {

    }

    /**
     * 客户端handler的初始化，服务器启动就执行
     */
    public void init() {
        WintcpSettings wintcpProperties = JavaImp.getInstance().getJimpSettings().getWintcp();
        port = wintcpProperties.getPort();
        peerport = wintcpProperties.getPeerport();
        peerModule = wintcpProperties.getPeerModule();
        ipaddress = wintcpProperties.getIpaddress();
        peerip = wintcpProperties.getPeerip();

        localpno = JavaImp.getInstance().getJimpSettings().getHandler().getPno();
        if (wintcpProperties.getByteOrder() != null) {
            byteOrder = wintcpProperties.getByteOrder() == 0 ? ByteOrder.BIG_ENDIAN : ByteOrder.LITTLE_ENDIAN;
        } else {
            byteOrder = ByteOrder.nativeOrder(); // LITTLE_ENDIAN
        }
    }

    public int getPeerModule() {
        return peerModule;
    }

    // 重新计划检测任务
    private void reScheduleCheck() {
        if (checkFuture != null) { // return;
            checkFuture.cancel(true);// 中断线程池中的任务
            checkFuture = null;
        }

        /**
         * Creates and executes a one-shot action that becomes enabled after the given
         * delay.
         * Parameters:command the task to executedelay the time from now to delay
         * executionunit the time unit of the delay parameter
         */ // 启动一个执行一次的线程（实现Runnable接口的类作为target） target，延迟时间数，时间单位
        checkFuture = checkSchedule.schedule(new CheckConnection(this), Const.TIME_IDLE, TimeUnit.MILLISECONDS);
    }

    // 重新连接，每次reconnect的间隔时间越来越长，100ms，200ms，300ms。。。
    private void reConnect() {
        long retrytime = Const.TIME_IDLE / 5;
        if (reconnectcount > MAX_RECONNECTCOUNT) {
            retrytime *= reconnectcount / MAX_RECONNECTCOUNT;
        }
        reConnect(retrytime > 5000 ? 5000 : retrytime);
    }

    private void reConnect(long waittime) {
        log.info(MessageFormat.format("Reschedule reconnect... to {0}", getConnectionStr()));
        if (connFuture != null) {
            connFuture.cancel(true);
            connFuture = null;
        }
        if (checkFuture != null) {
            checkFuture.cancel(true);
            checkFuture = null;
        }
        log.info(MessageFormat.format("Reconnect to {0} will start after {1} ms, retry {2} times", getConnectionStr(),
                waittime, reconnectcount));
        connFuture = connSchedule.schedule(this, waittime, TimeUnit.MILLISECONDS);
    }

    public void close() {
        if (checkFuture != null) {
            checkFuture.cancel(true);
            checkFuture = null;
        }
        if (socket != null) {
            try {
                log.info("Close client socket " + getConnectionStr());
                socket.close();
            } catch (Exception e) {
                log.info("Close client socket for [" + getConnectionStr()
                        + "] failed, maybe connection is closed.");
            }
        }
    }

    /**
     * peerip:peerport
     */
    public String getConnectionStr() {
        return (new StringBuffer()).append(peerip).append(":").append(peerport).toString();
    }

    /**
     * 客户端线程开始执行，
     */

    public synchronized void run() {
        close(); // 重新连接时会关闭socket和清空connFuture和checkFuture
        reconnectcount++;
        socket = null;
        // 创建连接，实现 IP 套接字地址（IP 地址 + 端口号），不依赖任何协议。
        SocketAddress remote = new InetSocketAddress(peerip, peerport);
        SocketAddress bindpoint = new InetSocketAddress(ipaddress, port);
        try {
            socket = new Socket();
            if (port > 0)
                socket.bind(bindpoint);
            else
                port = socket.getLocalPort(); // 若没有port则本地随机选择一个本地可用port

            log.info("Try to connect " + getConnectionStr() + " ...");
            socket.connect(remote, CONNECT_TIMEOUT); // 建立连接
            log.info("Try to connect " + getConnectionStr() + " socket connected");
            // 连接检测，立即进行（单次）
            byte[] checkByte = MessageHead.getCheckMessage(
                    (byte) JavaImp.getInstance().getJimpSettings().getWintcp().getModule().intValue(),
                    (byte) peerModule);
            
            reconnectcount = 0;
            try {
                log.info("try send hand shake to module: " + peerModule);
                socket.getOutputStream().write(checkByte);
                log.info("send hand shake to module: " + peerModule + " succ");
                reScheduleCheck(); // 开启检测线程（checkSchedule）重复检测是否处于连接状态
            } catch (Exception e) {
                log.info("hand shake exception: " + e.getMessage());
                reConnect();
                return;
            }

            log.info(MessageFormat.format("Connect to {0} successful.", getConnectionStr()));

            regDtmfFuture = regDtmfSchedule.schedule(new RegDtmfObserver(this), Const.TIME_REG_DTMF,
                    TimeUnit.MILLISECONDS);

            InputStream in = socket.getInputStream(); // 得到从客户端发送到服务器（本机）的数据流

            while (true) {
                if (!socket.isConnected()) {
                    log.info(MessageFormat.format("Connection for {0} isnot connected. Try reconnect...",
                            getConnectionStr()));
                    reConnect();
                    return;
                }
                byte[] bs = new byte[Const.MSG_HEAD_LEN];
                in.read(bs); // 读出消息头
                // reScheduleCheck();
                MessageHead head = new MessageHead();
                head.parse(bs, byteOrder); // 将消息头相关数据放入head
                // log.info(MessageFormat.format(
                // "Recv message from {0} [{1}.{2}=>{3}.{4}]?", peerip,
                // head.getSender().getModule(), head.getSender().getPno(),
                // head.getReceiver().getModule(),
                // head.getReceiver().getPno()));
                // 合法性检测
                if (!head.isValid()) {
                    log.error(MessageFormat.format(
                            "Recv invalid message from {0} [{1}.{2}=>{3}.{4}]?", peerip,
                            head.getSender().getModule(), head.getSender().getPno(), head.getReceiver().getModule(),
                            head.getReceiver().getPno()));
                    // log.debug(MD5.byte2HexStr(head.getByteArray(ByteOrder.BIG_ENDIAN)));
                    // reConnect(1); //合法性检测失败重新连接，开启下个线程
                    // return;//本线程结束
                    continue; // zwan: 不重连，继续
                } else if (!head.isCheckMessage()
                        && head.getReceiver().getModule() != (short) JavaImp.getInstance().getJimpSettings().getWintcp()
                                .getModule().intValue())
                // 此包不是发给本模块的
                {
                    log.error("Recived message module is " +
                            head.getReceiver().getModule() + " but not me");
                    // // 实现转发功能
                    // byte[] bs1 = new byte[head.getBodyLen()];
                    // in.read(bs1);
                    // JavaImp.getInstance().sendMessage(head, bs1);
                    continue;
                }

                if (head.isCheckMessage() || head.getReceiver().getPno() == 0) // 检测包，直接跳过
                {
                    // log.info(">>>>>>>>>>收到SCP发送来的检测包，包中head中receiver的pno=" + head.getReceiver().getPno());
                    if (head.getBodyLen() > 0) {
                        byte[] bs1 = new byte[head.getBodyLen()];
                        in.read(bs1);
                    }
                    // 连接检测包，不返回
                    // ByteBuffer bf1 =
                    // MessageHead.getCheckMessageBuff(head.getReceiver().getModule(),
                    // (short)JavaImp.getInstance().getModuleNo());
                    // bf1.order(ByteOrder.BIG_ENDIAN);
                    // bf1.rewind();
                    // socket.getOutputStream().write(bf1.array());
                } else {
                    // 正式包
                    // 正式的数据包
                    byte[] bs1 = new byte[head.getBodyLen()];
                    int readCount = 0;
                    while (readCount < head.getBodyLen()) {
                        readCount += in.read(bs1, readCount, head.getBodyLen() - readCount);
                    }
                    // 只有在这里才能保证顺序，否则提交到线程组里面
                    if (head.getEventid() == Const.EV_JANUS_TO_OAM_ROOM_MESSAGE) {
                        RtcmsResponse response = new RtcmsResponse();
                        ByteBuffer buffer = ByteBuffer.wrap(bs1);
                        buffer.order(byteOrder);
                        buffer.rewind();
                        response.readFromBuffer(buffer);
                        log.info("got rtcgw status message:" + response.getBody());
                        JSONObject body = JSONObject.parseObject(response.getBody());
                        JSONObject data = body.getJSONObject("data");
                        String room = data.getString("room");

                        // JavaImp jimp = JavaImp.getInstance();
                        // UserStatusHandler handler =
                        // (UserStatusHandler)jimp.getHandlerInstance(UserStatusHandler.class);

                        String ob = UserStatusHandler.getRoomOberver(room);
                        if (ob == null)
                            continue;

                        try {
                            HttpRequest httpRequest = HttpRequest.post(ob, null, Boolean.TRUE);
                            httpRequest.connectTimeout(3000);
                            httpRequest.readTimeout(4000);
                            httpRequest.contentType("application/json", "UTF-8");
                            if (ob.startsWith("https")) {
                                httpRequest.trustAllCerts();
                                httpRequest.trustAllHosts();
                            }
                            httpRequest.send(response.getBody());

                            if (httpRequest.code() != 200) {
                                log.info("callback url:" + ob + " failed! ret code: " + httpRequest.code());
                            }
                        } catch (Exception ex) {
                            log.info("send to url " + ob + " error! " + ex.getMessage());
                            continue;
                        }
                        continue;
                    }

                    JavaImp.getInstance().getExecutor().submit(
                            new HandlerRunner(head, ByteBuffer.wrap(bs1), JavaImp.getInstance().getByteOrder()));
                }
            }
        } catch (Exception e) {
            log.info(MessageFormat.format("Connect to {0}:{1,number,#} failed: {2}", peerip, peerport, e.getMessage()));

            if (socket != null) {
                try {
                    socket.close();
                    socket = null;
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }

            if (checkFuture == null) {
                checkFuture = checkSchedule.schedule(new CheckConnection(this), Const.TIME_IDLE, TimeUnit.MILLISECONDS);
            }
            reConnect();
        }
    }

    public void sendMessage(int srcpno, int descpno, int eventid, MsgStruct msg) {
        sendMessage(srcpno, descpno, eventid, msg.getByteArray(byteOrder));
    }

    public void sendMessage(int srcpno, int descpno, int eventid, byte[] body) {
        MessageHead head = new MessageHead();
        head.getSender().setModule((short) JavaImp.getInstance().getModuleNo());
        head.getSender().setPno(srcpno);
        head.getReceiver().setModule((short) peerModule);
        head.getReceiver().setPno(descpno);
        head.setBodyLen(body.length);
        head.setSecretfield(head.calcSecretField());
        sendMessage(head, body);
    }

    public void sendMessage(MessageHead head, byte[] body) {
        ByteBuffer bf = ByteBuffer.allocate(head.getLength() + body.length);
        head.writeToBuffer(bf);
        bf.put(body);
        bf.rewind();

        try {
            socket.getOutputStream().write(bf.array());
        } catch (Exception e) {
            log.info(MessageFormat.format("Send message to module [{0}] error, eventid: {1,number,#}, length: {2}",
                    head.getReceiver().getModule(), head.getEventid(), body.length));
            // reConnect(); //发送失败不能reConnect，否则会不断schedule重试
        }
    }

    public Socket getSocket() {
        return this.socket;
    }

    public ByteOrder getByteOrder() {
        return byteOrder;
    }

    // public int getTermno(){
    // return termno;
    // }

    @Override
    protected void finalize() throws Throwable {
        super.finalize();
    }

    class CheckConnection implements Runnable {
        private ClientHandler h;

        public CheckConnection(ClientHandler h) {
            this.h = h;
        }

        public synchronized void run() {
            byte[] bs = MessageHead.getCheckMessage((byte) JavaImp.getInstance().getModuleNo(), (byte) peerModule);
            try {
                socket.getOutputStream().write(bs); // 每5s发送一次测试数据，检查是否处于连接状态，若socket连接失败，则抛出异常
                // System.out.println("<<<<<<<<<<<<<向SCP发送检测数据，测试是否连接");
                // log.info("send  ====  " + socket.getInetAddress().getHostAddress() + "  " + bs.length);
                if (checkFuture != null) { // 保证检测线程只有一条
                    checkFuture.cancel(true);
                    checkFuture = null;
                }
                checkFuture = checkSchedule.schedule(new CheckConnection(h), Const.TIME_IDLE, TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                if (checkFuture != null) { // 结束检测，清空检测线程池
                    checkFuture.cancel(true);
                    checkFuture = null;
                }
                h.close();
                this.h = null;
                log.info(MessageFormat.format("Send check message to {0} error : ", socket.getInetAddress()
                        .getHostAddress(), e.getMessage()));
            }
        }
    }

    class RegDtmfObserver implements Runnable {
        private ClientHandler h;

        public RegDtmfObserver(ClientHandler h) {
            this.h = h;
        }

        public synchronized void run() {
            MessageHead head = new MessageHead();
            head.setEventid(Const.EV_JANUS_CONNECTED_MESSAGE);
            head.setDestPno(localpno);
            JavaImp.getInstance().getExecutor().submit(
                    new HandlerRunner(head, null, JavaImp.getInstance().getByteOrder()));
        }
    }

}
